package net.bwie.realtime.jtp.ods.job;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.common.utils.MysqlCdcUtil;
import net.bwie.realtime.jtp.ods.function.CdcDataEtlMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 使用Flink从MySQL中读取jtp_mall数据库数据，并且进行ETL转换操作，储存到Kafka中
 */
public class JtpCdcMallDataJob {
    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);
        //获取数据源
        DataStream<String> dataStream = cdcMysqlSource(env);

        //转换数据
        DataStream<String> etlStream = dataStream.map(new CdcDataEtlMapFunction());
//        etlStream.print();
        //发送到kafka

        KafkaUtil.producerKafka(etlStream, "topic-db");
        env.execute("JtpCdcMallDataJob");


    }

    private static DataStream<String> cdcMysqlSource(StreamExecutionEnvironment env) {
        //todo 全量+增量
        String[] dimList=new String[]{"base_dic","base_province","base_region","base_category1","base_category2","base_category3",
        "activity_info","activity_rule","base_trademark","coupon_info","sku_info","spu_info","user_info"
        };
        DataStream<String> dataStream = MysqlCdcUtil.cdcMysqlInitial(env, "jtp_mall", dimList);
        dataStream.print("data");
        //todo 增量
        String[] factList=new String[]{"favor_info","cart_info","order_info","order_detail","order_detail_activity","order_detail_coupon",
                "order_status_log","payment_info","order_refund_info","refund_payment","coupon_info","coupon_use"
        };
        DataStream<String> cdcFactStream = MysqlCdcUtil.cdcMysqlEarliest(env, "jtp_mall", factList);
        cdcFactStream.print("cdc");
        DataStream<String> union = dataStream.union(cdcFactStream);
        return union;
    }
}
